
package org.codebase.hadoop.zookeeper.cluster;

import java.io.IOException;
import java.util.List;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

/**
 * 监视器
 **************************************************
 * ClassName:ClusterMonito
 * 

 * @author:xiaozhichao
 * @date:2018年8月17日
 **************************************************
 */
public class ClusterMonito implements Runnable {

	private final Watcher connectionWatcher;
	private final Watcher childrenWatcher;
	private ZooKeeper zk;
	@SuppressWarnings("unused")
	private String root;
	private boolean alive = true;

	public ClusterMonito(String hostPort, String root) throws IOException, KeeperException, InterruptedException {
		this.root = root;
		this.connectionWatcher = new Watcher() {

			@Override
			public void process(WatchedEvent event) {
				if (event.getType() == Watcher.Event.EventType.None
						&& event.getState() == Watcher.Event.KeeperState.SyncConnected) {
					System.out.printf("\nEvent Received: %s", event.toString());
				}
			}

		};
		this.childrenWatcher = new Watcher() {

			@Override
			public void process(WatchedEvent event) {
				System.out.printf("\nEvent Received: %s", event.toString());
				if (event.getType() == Watcher.Event.EventType.NodeChildrenChanged) {
					try {
						List<String> children = zk.getChildren(root, this);
						System.out.printf("\nMESSAGE: %s", "!!!Cluster Membership Change!!!");
						System.out.printf("\nMESSAGE: %s", "Members: " + children);
					} catch (KeeperException e) {
						throw new RuntimeException(e);

					} catch (InterruptedException e) {
						Thread.currentThread().interrupt();
						alive = false;
						throw new RuntimeException(e);

					}
				}

			}

		};
		this.zk = new ZooKeeper(hostPort, 2000, connectionWatcher);
		if (zk.exists(root, false) == null) {
			zk.create(root, "ClusterMonitorRoot".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
		}

		List<String> children = zk.getChildren(root, childrenWatcher);
		System.err.println("Members: " + children);

	}

	@Override
	public void run() {

		try {
			synchronized (this) {
				while (this.alive) {
					wait();
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
			Thread.currentThread().interrupt();
		} finally {
			try {
				this.zk.close();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	public boolean isAlive() {

		return this.alive;
	}

	public void setAlive(boolean alive) {

		this.alive = alive;
	}

}
